package com.weizhu.swufer.core.rpc;

import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.weizhu.swufer.constant.SwuferConstant;
import com.weizhu.swufer.core.exception.RpcException;
import com.weizhu.swufer.core.util.ProtoUtil;
import com.weizhu.swufer.grpc.Response;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class RpcFuture<T> implements ListenableFuture<T> {

    private ListenableFuture<Response> listenableFuture;

    public RpcFuture(ListenableFuture<Response> listenableFuture){
        this.listenableFuture = listenableFuture;
    }


    public void addCallback(FutureCallback<T> callback,Executor executor){

        Futures.addCallback(this,callback,executor);
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return listenableFuture.cancel(mayInterruptIfRunning);
    }

    @Override
    public boolean isCancelled() {
        return listenableFuture.isCancelled();
    }

    @Override
    public boolean isDone() {
        return listenableFuture.isDone();
    }

    @Override
    public T get() throws InterruptedException, ExecutionException {
        Response response;
        RpcResponse rpcResponse;
        try {
            response = listenableFuture.get();
        }catch (Exception e){
            throw new RpcException(e.getMessage(),e);
        }
        rpcResponse = ProtoUtil.deserialize(response.getResponse().toByteArray(), RpcResponse.class);
        return getDataFromRpcResponse(rpcResponse);
    }

    @Override
    public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        Response response;
        RpcResponse rpcResponse;
        try {
            response = listenableFuture.get(timeout, unit);
        }catch (Exception e){
            throw new RpcException(e.getMessage(),e);
        }
        rpcResponse = ProtoUtil.deserialize(response.getResponse().toByteArray(), RpcResponse.class);
        return getDataFromRpcResponse(rpcResponse);
    }

    @SuppressWarnings("unchecked")
    private T getDataFromRpcResponse(RpcResponse rpcResponse){
        if (rpcResponse.getCode() != SwuferConstant.RESPONSE_SUCCESS_CODE){
            Throwable remoteError = rpcResponse.getError();
            if (remoteError instanceof RpcException){
                throw (RpcException)remoteError;
            }
            throw new RpcException(rpcResponse.getMsg(), remoteError);
        }

        return (T) rpcResponse.getData();
    }


    @Override
    public void addListener(Runnable runnable, Executor executor) {

    }
}
